package job;

import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import javax.persistence.EntityManager;
import javax.persistence.EntityTransaction;

import models.SellUnknownOrder;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import kafka.serializer.StringDecoder;
import play.Logger;
import play.db.jpa.JPA;
import play.jobs.Every;
import play.jobs.Job;
import utils.DateUtils;
import utils.PropertyUtil;
import utils.TopicUtil;
import bp.NetSaleBP;

/**
 * 
 * <b>类描述：售票消费者</b><br/>
 * <b>类名称：</b>UnlockConsumer<br/>
 * <b>创建人：</b>郭子义</a><br/>
 * <b>关键修改：</b><br/>
 * <b>修改时间：</b><br/>
 * <b>修改人：</b><br/>
 * 
 */
@SuppressWarnings("rawtypes")
@Every("10s")
public class SellUnknownConsumer extends Job {
//	private final ConsumerConnector consumer;
//	private final String topic;

//	public SellUnknownConsumer() {
//
//		System.out.println("before create consumer");
//		consumer = kafka.consumer.Consumer.createJavaConsumerConnector(createConsumerConfig());// 创建到zookeeper的连接
//		System.out.println("after create consumer");
//		this.topic = TopicUtil.getTopic("SellUnknown");
//	}

//	private ConsumerConfig createConsumerConfig() {
//		Properties props = new Properties();
//		props.put("zookeeper.connect", KafkaProperties.zkConnect);// 要连接的zookeeper地址
//		props.put("group.id", KafkaProperties.groupId);// 消费者所在的组
////		props.put("conusmeor.id", KafkaProperties.consumerId);// 此消费者的唯一标识用于在zookeeper中存储此消费者的偏移量
//		//props.put("conusmeor.id", "SellUnknownConsumer");// 此消费者的唯一标识用于在zookeeper中存储此消费者的偏移量
//		props.put("zookeeper.session.timeout.ms", KafkaProperties.zSessionTimeout);// zookeeper连接超时时间
//		// props.put("zookeeper.sync.time.ms", "2000");
//		props.put("auto.offset.reset", "smallest");
//		props.put("auto.commit.enable", "false");
//
//		return new ConsumerConfig(props);
//	}

	@Override
	public void doJob() throws Exception {
		int maxconsumercount = Integer.parseInt(PropertyUtil.getProperty("maxconsumercount"));
		NetSaleBP bp = new NetSaleBP();
		List<SellUnknownOrder> unknownorders = JPA.em().createQuery("from SellUnknownOrder where dealed=0 and dealcount<="+maxconsumercount, SellUnknownOrder.class).getResultList();
		try {
			for(SellUnknownOrder order : unknownorders){
				try {
					Date msgtime=order.msgtime;
					Date excutetime = DateUtils.add(msgtime, Calendar.MINUTE, (int)Math.pow(20, 0.5*order.dealcount));
					Date now = new Date();
					
					if(excutetime.before(now)){//到了执行时间
						Logger.log4j.info("SellUnknownOrder第三态售票开始处理，订单号："+order.orderno+",失败次数："+order.dealcount+",excutetime"+excutetime);
						boolean flag = false;
						try {
							flag = bp.sellunknown(order.msgcontent);
						} catch (Exception e) {
							flag = false;
						}
						if(flag){
							
							order.dealed = true;
							order.dealtime = now;
						}else{
							order.dealcount = order.dealcount + 1;
						}
						if(!JPA.em().getTransaction().isActive()){
							JPA.em().getTransaction().begin();
						}
						order.save();
						JPA.em().getTransaction().commit();
					}else{
						continue;
					}
				} catch (Exception e) {
					Logger.log4j.error("SellUnknownOrder第三态售票处理失败，订单号："+order.orderno+",失败次数："+order.dealcount, e);
				}
			}

		} catch (Exception e) {
			Logger.log4j.error("SellUnknownOrder第三态售票处理失败：", e);
		}
	}

	public static void main(String[] args) {
		for(int i=1;i<6;i++){
			System.out.println((int)Math.pow(20, 0.5*i));
		}
	}
}
